package com.young.trigger.handler;

import com.young.common.InetTool;
import com.young.common.query.Query;
import com.young.common.query.QueryBuilder;
import com.young.core.domain.JobDO;
import com.young.core.domain.TriggerDO;
import com.young.core.enums.LockTypeEnum;
import com.young.core.manager.BrokerManager;
import com.young.core.manager.JobManager;
import com.young.core.manager.TriggerManager;
import com.young.trigger.event.BrokerEvent;
import org.quartz.*;
import org.quartz.impl.matchers.GroupMatcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @description：
 * @author: yangyahui01
 * @date: 11/28/24 4:09 PM
 */
@Component
public class MasterHandler implements ApplicationListener<BrokerEvent>, Closeable {

    private static final Logger LOGGER = LoggerFactory.getLogger(MasterHandler.class);

    private final ExecutorService executor = Executors.newSingleThreadExecutor();

    private static final String GROUP = "_G_SCHEDULER_";



    @Autowired
    private Scheduler scheduler;

    @Autowired
    private JobManager jobManager;

    @Override
    public void onApplicationEvent(BrokerEvent event) {

        LockTypeEnum typeEnum = event.getSource();

        //如果此节点被提升为master，则首先清理本地的所有任务，然后重新启动
        if (typeEnum == LockTypeEnum.PROMOTED) {
            LOGGER.info("Master - 节点被选举为master,{}", InetTool.LOCAL_IP);

            executor.execute(() -> {
                stopAll();
                startAll();
            });
        } else if (typeEnum == LockTypeEnum.LEAVE) {
            LOGGER.info("Master - 节点切换为Slave,{}",InetTool.LOCAL_IP);

            executor.execute(() -> {
                stopAll();
            });
        }
    }

    private synchronized void startAll() {

        try {
            long begin = System.currentTimeMillis();

            while (BrokerContext.isLeading()) {
                LOGGER.info("Master - load任务列表开始...");

                Collection<JobDO> jobs = load();

                LOGGER.info("Master - load任务列表成功!, size:{}", jobs.size());

                if (jobs != null) {
                    for (JobDO job : jobs) {
                        LOGGER.info("start job:{}", job.getId());
                        start(job);//启动任务
                    }
                    LOGGER.info("Master - 所有任务启动成功,耗时 - {}ms",System.currentTimeMillis() - begin);
                    return;
                }
                Thread.sleep(200);
            }
        } catch (InterruptedException e) {
            LOGGER.info("Master - load任务中断");
        }
        LOGGER.error("Master - load任务失败,重试失败,master - {},local - {}",BrokerContext.isMaster(),BrokerContext.isLocal());
    }

    /**
     * 启动quartz中任务
     */
    private void start(JobDO job) {

        try {
            JobKey jobKey = new JobKey(job.getId() + "/" + job.getUniqueCode(), GROUP);

            if (scheduler.checkExists(jobKey)) {
                scheduler.deleteJob(jobKey);
            }

            LOGGER.info("jobId:{}", job.getId());

            JobDetail jobDetail = JobBuilder.newJob(QuartzJobFactory.class).withIdentity(jobKey)
                    .usingJobData("jobId", job.getId())
                    .usingJobData("project", job.getProject())
                    .usingJobData("version", job.getUpdateTime().getTime()).build();


            CronScheduleBuilder builder = CronScheduleBuilder.cronSchedule(job.getExpression());

            TriggerKey triggerKey = new TriggerKey(job.getId() + "/" + job.getUniqueCode(), GROUP);

            Trigger trigger = TriggerBuilder.newTrigger()
                    .withIdentity(triggerKey)
                    .withSchedule(builder)
                    .startNow()
                    .build();

            scheduler.scheduleJob(jobDetail, trigger);
        } catch (SchedulerException e) {
            LOGGER.error("Master - 启动任务失败,{}", job.getUniqueCode(), e);
        }
    }

    /**
     * 查询所有任务
     * @return
     */
    private Collection<JobDO> load() {
        int i = 1;
        try {
            Collection<JobDO> jobs = new ArrayList<>();
            while (BrokerContext.isLeading()) {
                Query query = QueryBuilder.page(i).pageSize(128).put("status", 1).build();
                Collection<JobDO> result = jobManager.query(query).getResult();
                if (result.isEmpty()) {
                    return jobs;
                }
                jobs.addAll(result);
                ++i;
            }
        } catch (Exception e) {
            LOGGER.error("Master - load数据库查询数据失败", e);
        }
        return null;
    }

    /**
     * 关闭所有任务
     */
    private void stopAll() {
        try {
            Set<JobKey> jobKeys = scheduler.getJobKeys(GroupMatcher.groupEquals(GROUP));
            for (JobKey jobKey : jobKeys) {
                scheduler.deleteJob(jobKey);
            }
            LOGGER.info("Master - 任务全部停止成功");
        } catch (SchedulerException e) {
            LOGGER.error("Master - 关闭任务失败", e);
        }
    }

    @Override
    public void close() throws IOException {

    }


    /**
     * 任务执行类
     * 这里只是把任务分配给具体的 trigger broker 节点，并没有真正的去请求业务端
     */
    public static class QuartzJobFactory implements Job {

        @Autowired
        private TriggerManager triggerManager;

        @Autowired
        private BrokerManager brokerManager;

        @Override
        public void execute(JobExecutionContext context) throws JobExecutionException {

            TriggerKey key = context.getTrigger().getKey();
            LOGGER.debug("Master - quartz任务触发,key - {}", key);

            if (!BrokerContext.isLeading()) {
                LOGGER.warn("Master - quartz任务触发,但当前master已迁移,担保执行,key - ", key);
            }

            TriggerDO trigger = new TriggerDO();

            try {
                JobDataMap data = context.getMergedJobDataMap();

                int jobId = data.getInt("jobId");
                String project = data.getString("project");

                List<String> ips = brokerManager.getCache();
                String ip = choose(new ArrayList<>(ips), jobId);
                trigger.setJobId(jobId);
                trigger.setProject(project);
                trigger.setIp(ip);
                trigger.setTs(System.currentTimeMillis());

                triggerManager.insert(trigger);

                LOGGER.info("Master - quartz新增trigger任务：key:{}, triggerId:{}", key, trigger.getId());
            } catch (Exception e) {
                LOGGER.error("Master - quartz任务执行失败,key - {}", key,e);
//                slaveHandler.submit(trigger);//逃生
            }


        }
    }

    /**
     * 从broker列表中，选择执行job的broker地址
     * @param ips
     * @param jobId
     * @return
     */
    private static String choose(List<String> ips, int jobId) {
        ips.removeIf(ip -> ip.equals(InetTool.LOCAL_IP));
        //node列表为空时，master承担执行
        if (ips.isEmpty()) {
            return InetTool.LOCAL_IP;
        }
        return ips.get(jobId % ips.size());
    }
}
